package articleEvents

import (
	"context"
	"github.com/IBM/sarama"
	"go.uber.org/zap"
	"jk-time/webook/interactive/repository"
	"jk-time/webook/pkg/saramax"
	"time"
)

type InteractiveKafkaComsumer struct {
	client sarama.Client
	repo   repository.InteractiveRepository
}

func NewInteractiveKafkaComsumer(client sarama.Client, repo repository.InteractiveRepository) *InteractiveKafkaComsumer {
	return &InteractiveKafkaComsumer{
		client: client,
		repo:   repo,
	}
}
func (k *InteractiveKafkaComsumer) Start() error {
	cg, err := sarama.NewConsumerGroupFromClient("histroy", k.client)
	if err != nil {
		return err
	}
	go func() {
		err := cg.Consume(context.Background(),
			[]string{ReadEvent{}.Topic()},
			saramax.NewHandler[ReadEvent]("histroy", k.Consume))
		if err != nil {
			zap.L().Error("退出了消费循环,异常", zap.Error(err))
		}
	}()
	return err
}

func (k *InteractiveKafkaComsumer) StartBatch() error {
	cg, err := sarama.NewConsumerGroupFromClient("histroy", k.client)
	if err != nil {
		return err
	}
	go func() {
		err := cg.Consume(context.Background(),
			[]string{ReadEvent{}.Topic()},
			saramax.NewHandler[ReadEvent]("histroy", k.Consume))
		if err != nil {
			zap.L().Error("退出了消费循环,异常", zap.Error(err))
		}
	}()
	return err
}
func (k *InteractiveKafkaComsumer) Consume(msg *sarama.ConsumerMessage, t ReadEvent) error {
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	//mysql+redis
	return k.repo.IncrReadCnt(ctx, "article", t.Aid)
}

func (k *InteractiveKafkaComsumer) BatchConsume(msgs []*sarama.ConsumerMessage, ts []ReadEvent) error {
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()

	bizs := make([]string, 0, len(msgs))
	ids := make([]int64, 0, len(ts))
	for _, evt := range ts {
		bizs = append(bizs, "article")
		ids = append(ids, evt.Uid)
	}
	//tsByte, _ := json.Marshal(ts)
	//zap.L().Info("kafka:", zap.String("topic", ReadEvent{}.Topic()), zap.ByteString("value", tsByte))
	//mysql+redis
	return k.repo.BatchIncrReadCnt(ctx, bizs, ids)
}
